media_pp\core\pipeline/
chain.rs

1use std::sync::Arc;
2
3use crate::pp_log::{PpLog, pp_trace};
4
5use crate::{
6    buffer::MediaBuffer,
7    bus::{Bus, BusEvent},
8    control::ControlMsg,
9    element::{Context, Element, ElementType, Filter, Sink, Source, element_pp_log},
10    error::Result,
11    graph::{BranchId, BranchPlan, ElementId, GraphError, NodeInfo, PlannedEdge, PortRef},
12    pad::SrcPad,
13    queue::{OverflowPolicy, Queue},
14};
15
16/// Builds one chain segment (a run of elements that all execute on the same
17/// thread). Call [`ChainBuilder::queue`] to close the current segment behind
18/// a `Queue` and start a new one on its own worker thread.
19///
20/// Because each element needs a handle to *its* downstream to be
21/// constructed, the chain is assembled back-to-front: elements are
22/// collected in call order, then folded right-to-left starting from the
23/// terminal `Sink` at [`ChainBuilder::to`] time.
24pub struct ChainBuilder {
25    context: Arc<Context>,
26    elements: Vec<Box<dyn StageBuilder>>,
27    /// Nodes kept locally until this builder becomes a `DetachedBranch`
28    /// and an attach operation commits the complete plan.
29    planned: Vec<PlannedNode>,
30    error: Option<GraphError>,
31}
32
33struct PlannedNode {
34    info: NodeInfo,
35    output_port: Arc<str>,
36}
37
38/// A fully constructed runtime chain whose graph nodes are still detached.
39/// Dropping it has no topology effect; only an attach operation commits it.
40pub struct DetachedBranch {
41    pub(crate) root: Box<dyn Sink>,
42    pub(crate) plan: BranchPlan,
43}
44
45impl DetachedBranch {
46    pub fn root_id(&self) -> ElementId {
47        self.plan.root
48    }
49}
50
51trait StageBuilder: Send {
52    fn wrap(
53        self: Box<Self>,
54        downstream: Box<dyn Sink>,
55        bus: &Bus,
56        pipeline_id: &str,
57    ) -> Box<dyn Sink>;
58}
59
60struct DirectStage<T>(T);
61
62/// Adds uniform EOS/control boundary tracing to every direct filter without
63/// requiring each built-in or downstream custom element to duplicate it.
64struct FlowTracer<T> {
65    inner: T,
66}
67
68impl<T: Element> Element for FlowTracer<T> {
69    fn name(&self) -> Arc<str> {
70        self.inner.name()
71    }
72
73    fn element_type(&self) -> ElementType {
74        self.inner.element_type()
75    }
76
77    fn graph_id(&self) -> Option<ElementId> {
78        self.inner.graph_id()
79    }
80
81    fn pp_log(&self) -> &PpLog {
82        self.inner.pp_log()
83    }
84
85    fn pp_log_mut(&mut self) -> &mut PpLog {
86        self.inner.pp_log_mut()
87    }
88}
89
90impl<T: Source> Source for FlowTracer<T> {
91    fn src_pads(&mut self) -> &mut [SrcPad] {
92        self.inner.src_pads()
93    }
94}
95
96impl<T: Sink> Sink for FlowTracer<T> {
97    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
98        let is_eos = buf.is_eos();
99        if is_eos {
100            pp_trace!(pp_log: self.inner.pp_log(), "event=eos phase=received");
101        }
102        let result = self.inner.consume(buf);
103        if is_eos {
104            match &result {
105                Ok(()) => pp_trace!(
106                    pp_log: self.inner.pp_log(),
107                    "event=eos phase=completed outcome=ok"
108                ),
109                Err(error) => pp_trace!(
110                    pp_log: self.inner.pp_log(),
111                    "event=eos phase=completed outcome=error error={error}"
112                ),
113            }
114        }
115        result
116    }
117
118    fn control(&mut self, msg: ControlMsg) -> Result<()> {
119        pp_trace!(
120            pp_log: self.inner.pp_log(),
121            "event=control control={msg:?} phase=received"
122        );
123        let result = self.inner.control(msg);
124        match &result {
125            Ok(()) => pp_trace!(
126                pp_log: self.inner.pp_log(),
127                "event=control control={msg:?} phase=completed outcome=ok"
128            ),
129            Err(error) => pp_trace!(
130                pp_log: self.inner.pp_log(),
131                "event=control control={msg:?} phase=completed outcome=error error={error}"
132            ),
133        }
134        result
135    }
136}
137
138impl<T> StageBuilder for DirectStage<T>
139where
140    T: Filter + 'static,
141{
142    fn wrap(
143        self: Box<Self>,
144        downstream: Box<dyn Sink>,
145        _bus: &Bus,
146        pipeline_id: &str,
147    ) -> Box<dyn Sink> {
148        let mut element = self.0;
149        *element.pp_log_mut() =
150            element_pp_log(element.element_type(), &element.name(), Some(pipeline_id));
151        element.src_pads()[0].link(downstream);
152        Box::new(FlowTracer { inner: element })
153    }
154}
155
156struct QueueStage {
157    id: ElementId,
158    name: String,
159    capacity: usize,
160    policy: OverflowPolicy,
161}
162
163/// Traces EOS/control at a terminal `Sink` and posts a `BusEvent::Eos` (under
164/// the sink's own `Element::name()`) once EOS completes — mirrors what
165/// `Queue` does for its own downstream, but without introducing a thread
166/// boundary. This is what lets a fully direct chain (no `queue()` calls at
167/// all) still report EOS on the bus.
168struct TerminalTracer {
169    bus: Bus,
170    inner: Box<dyn Sink>,
171}
172
173impl Element for TerminalTracer {
174    fn name(&self) -> Arc<str> {
175        self.inner.name()
176    }
177
178    fn element_type(&self) -> ElementType {
179        self.inner.element_type()
180    }
181
182    fn graph_id(&self) -> Option<ElementId> {
183        self.inner.graph_id()
184    }
185
186    fn pp_log(&self) -> &PpLog {
187        self.inner.pp_log()
188    }
189
190    fn pp_log_mut(&mut self) -> &mut PpLog {
191        self.inner.pp_log_mut()
192    }
193}
194
195impl Sink for TerminalTracer {
196    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
197        let is_eos = buf.is_eos();
198        if is_eos {
199            pp_trace!(pp_log: self.inner.pp_log(), "event=eos phase=received");
200        }
201        let result = self.inner.consume(buf);
202        if is_eos {
203            match &result {
204                Ok(()) => {
205                    pp_trace!(
206                        pp_log: self.inner.pp_log(),
207                        "event=eos phase=completed outcome=ok"
208                    );
209                    self.bus.post(
210                        self.inner.pp_log(),
211                        BusEvent::Eos {
212                            element_type: self.inner.element_type(),
213                            name: self.inner.name(),
214                        },
215                    );
216                }
217                Err(error) => pp_trace!(
218                    pp_log: self.inner.pp_log(),
219                    "event=eos phase=completed outcome=error error={error}"
220                ),
221            }
222        }
223        result
224    }
225
226    fn control(&mut self, msg: ControlMsg) -> Result<()> {
227        pp_trace!(
228            pp_log: self.inner.pp_log(),
229            "event=control control={msg:?} phase=received"
230        );
231        let result = self.inner.control(msg);
232        match &result {
233            Ok(()) => pp_trace!(
234                pp_log: self.inner.pp_log(),
235                "event=control control={msg:?} phase=completed outcome=ok"
236            ),
237            Err(error) => pp_trace!(
238                pp_log: self.inner.pp_log(),
239                "event=control control={msg:?} phase=completed outcome=error error={error}"
240            ),
241        }
242        result
243    }
244}
245
246impl StageBuilder for QueueStage {
247    fn wrap(
248        self: Box<Self>,
249        downstream: Box<dyn Sink>,
250        bus: &Bus,
251        pipeline_id: &str,
252    ) -> Box<dyn Sink> {
253        Box::new(Queue::spawn_with_policy(
254            self.name,
255            self.capacity,
256            downstream,
257            bus.for_element(self.id),
258            self.policy,
259            Some(pipeline_id),
260        ))
261    }
262}
263
264impl ChainBuilder {
265    /// Starts a detached branch plan. Prefer [`Context::branch`] at call
266    /// sites; it makes the owning pipeline explicit without cloning the
267    /// context manually.
268    pub fn new(context: Arc<Context>) -> Self {
269        Self {
270            context,
271            elements: Vec::new(),
272            planned: Vec::new(),
273            error: None,
274        }
275    }
276
277    /// Adds a single-output `Filter` (decoder, encoder, filter, ...) that
278    /// receives via `Sink` and produces through its own (single) src pad.
279    /// It runs on the same thread as whatever is upstream of it — direct
280    /// function call, no queue.
281    pub fn pipe<T: Filter + 'static>(mut self, mut element: T) -> Self {
282        let name = element.name();
283        let pad_count = element.src_pads().len();
284        if pad_count != 1 && self.error.is_none() {
285            self.error = Some(GraphError::NotSingleOutput {
286                name: name.clone(),
287                count: pad_count,
288            });
289        }
290        let output_port = element
291            .src_pads()
292            .first()
293            .map(|pad| Arc::<str>::from(pad.name()))
294            .unwrap_or_else(|| "src".into());
295        self.planned.push(PlannedNode {
296            info: NodeInfo {
297                id: self.context.graph.reserve_element_id(),
298                element_type: element.element_type(),
299                name,
300            },
301            output_port,
302        });
303        self.elements.push(Box::new(DirectStage(element)));
304        self
305    }
306
307    /// Introduces a thread boundary (blocking when full — see
308    /// [`OverflowPolicy::Block`]): everything added after this runs on its
309    /// own worker thread instead of the thread that feeds this queue.
310    pub fn queue(self, name: impl Into<String>, capacity: usize) -> Self {
311        self.queue_with_policy(name, capacity, OverflowPolicy::default())
312    }
313
314    /// Same as [`ChainBuilder::queue`], but lets you choose what happens
315    /// when the queue is full (e.g. [`OverflowPolicy::DropNewest`] for a
316    /// live source that shouldn't stall upstream).
317    pub fn queue_with_policy(
318        mut self,
319        name: impl Into<String>,
320        capacity: usize,
321        policy: OverflowPolicy,
322    ) -> Self {
323        let name: Arc<str> = name.into().into();
324        let id = self.context.graph.reserve_element_id();
325        self.planned.push(PlannedNode {
326            info: NodeInfo {
327                id,
328                element_type: ElementType::Queue,
329                name: name.clone(),
330            },
331            output_port: format!("{name}_src").into(),
332        });
333        self.elements.push(Box::new(QueueStage {
334            id,
335            name: name.to_string(),
336            capacity,
337            policy,
338        }));
339        self
340    }
341
342    /// Terminates the chain with a `Sink` (muxer, file sink, ...) and
343    /// assembles everything into a single `Box<dyn Sink>` ready to be
344    /// linked into a source's src pad. The terminal's own `Element::name()`
345    /// is what shows up on the bus when it reports EOS.
346    pub fn to(self, mut terminal: Box<dyn Sink>) -> Result<DetachedBranch> {
347        if let Some(error) = self.error {
348            return Err(error.into());
349        }
350        *terminal.pp_log_mut() = element_pp_log(
351            terminal.element_type(),
352            &terminal.name(),
353            Some(&self.context.pipeline_id),
354        );
355        let terminal_info = NodeInfo {
356            id: terminal
357                .graph_id()
358                .unwrap_or_else(|| self.context.graph.reserve_element_id()),
359            element_type: terminal.element_type(),
360            name: terminal.name(),
361        };
362        let terminal_id = terminal_info.id;
363        let mut nodes: Vec<_> = self.planned.iter().map(|node| node.info.clone()).collect();
364        nodes.push(terminal_info);
365        let edges = nodes
366            .windows(2)
367            .enumerate()
368            .map(|(index, pair)| PlannedEdge {
369                from: PortRef {
370                    element: pair[0].id,
371                    port: self.planned[index].output_port.clone(),
372                },
373                to: PortRef {
374                    element: pair[1].id,
375                    port: "sink".into(),
376                },
377            })
378            .collect();
379        let root_id = nodes.first().expect("terminal always supplies one node").id;
380        let terminal: Box<dyn Sink> = Box::new(TerminalTracer {
381            bus: self.context.bus.for_element(terminal_id),
382            inner: terminal,
383        });
384        let root = self
385            .elements
386            .into_iter()
387            .rev()
388            .fold(terminal, |downstream, stage| {
389                stage.wrap(downstream, &self.context.bus, &self.context.pipeline_id)
390            });
391        Ok(DetachedBranch {
392            root,
393            plan: BranchPlan {
394                nodes,
395                edges,
396                root: root_id,
397            },
398        })
399    }
400
401    pub fn build(self, terminal: Box<dyn Sink>) -> Result<DetachedBranch> {
402        self.to(terminal)
403    }
404}
405
406impl Context {
407    pub fn branch(self: &Arc<Self>) -> ChainBuilder {
408        ChainBuilder::new(self.clone())
409    }
410
411    pub fn attach<S: Source>(
412        &self,
413        source: &mut S,
414        pad_index: usize,
415        branch: DetachedBranch,
416    ) -> Result<BranchId> {
417        let pads = source.src_pads();
418        let pad_count = pads.len();
419        let pad = pads.get_mut(pad_index).ok_or(GraphError::PadOutOfRange {
420            index: pad_index,
421            pad_count,
422        })?;
423        self.attach_pad(pad, branch)
424    }
425
426    pub(crate) fn attach_pad(&self, pad: &mut SrcPad, branch: DetachedBranch) -> Result<BranchId> {
427        if pad.is_linked() {
428            return Err(GraphError::PadAlreadyLinked(pad.name().to_owned()).into());
429        }
430        let from_port: Arc<str> = pad.name().into();
431        let DetachedBranch { root, plan } = branch;
432        Ok(self
433            .graph
434            .attach_with(self.source_id, from_port, plan, |_| {
435                pad.link(root);
436                Ok(())
437            })?)
438    }
439}
media_pp\core\pipeline/chain.rs

media_pp\core\pipeline/
chain.rs
